package com.sam.kafka.producer;

import com.sam.kafka.entity.User;
import com.sam.kafka.serialization.UserSerializer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @author: wm.xue
 * @date: 2021/11/5 18:39
 * @description:
 */
public class _02_Producer_MySerialize {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.100.17.117:9092");
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put(ProducerConfig.ACKS_CONFIG, "-1");
        configs.put(ProducerConfig.RETRIES_CONFIG, 3);

        // 每次连接的最大请求数量, 设置重试失败,消息有序
        configs.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);
        // 设置自定义序列化器
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, UserSerializer.class);


        KafkaProducer<String, User> producer = new KafkaProducer<String, User>(configs);

        ProducerRecord<String, User> record
                = new ProducerRecord<String, User>("_01_serialize",0,"1",new User(1,"小明"));
        Future<RecordMetadata> future = producer.send(record);
        RecordMetadata recordMetadata = future.get();
        System.err.println(recordMetadata);


    }

}
